package com.jorado.pulsar.client.metrics;

import com.jorado.metrics.type.Gauge;
import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.impl.*;
import org.apache.pulsar.shade.org.eclipse.util.ConcurrentHashSet;

import java.util.Collections;
import java.util.Set;
import java.util.Timer;
import java.util.TimerTask;

/**
 *
 */
public class PulsarMetricsRecorder {

    private static long statsIntervalSeconds = 15;
    private static volatile PulsarMetricsRecorder pulsarMetricsRecorder = null;
    private final Set<ConsumerImpl<?>> consumersSet = new ConcurrentHashSet<>();
    private final Set<ProducerImpl<?>> producerSet = new ConcurrentHashSet<>();

    private boolean producerMetricsIsInit = false;
    private boolean consumerMetricsIsInit = false;

    private Timer timer;

    private static final String[][] CONSUMER_INDICATOR = {
            {"numAcksFailed", "numbers of acknowledge filed"},
            {"numAcksSent", "numbers of acknowledge sent"},
            {"numBytesReceived", "numbers of bytes received"},
            {"numMsgsReceived", "numbers of messages received"},
            {"numReceiveFailed", "numbers of receive failed"},
            {"rateBytesReceived", "numbers of bytes received"},
            {"rateMsgReceived", "rate of message received"},
            {"totalAcksFailed", "total fo acknowledge failed"},
            {"totalAcksSent", "total of acknowledge sent"},
            {"totalBytesReceived", "total of bytes received"},
            {"totalMsgsReceived", "total of messages received"},
            {"totalReceivedFailed", "total of received failed"}
    };

    public static String[][] getConsumerIndicator() {
        return CONSUMER_INDICATOR;
    }

    public static String[][] getProducerIndicator() {
        return PRODUCER_INDICATOR;
    }

    private static final String[][] PRODUCER_INDICATOR = {
            {"numAcksReceived", "numbers of acknowledge received"},
            {"numBytesSent", "numbers of bytes sent"},
            {"numMsgSent", "numbers of messages sent"},
            {"numSendFailed", "numbers of send failed"},
            {"sendBytesRate", "rate of send bytes"},
            {"sendLatencyMillis50pct", "send latency millis of 50 percent"},
            {"sendLatencyMillis75pct", "send latency millis of 75 percent"},
            {"sendLatencyMillis95pct", "send latency millis of 95 percent"},
            {"sendLatencyMillis99pct", "send latency millis of 99 percent"},
            {"sendLatencyMillis999pct", "send latency millis of 999 percent"},
            {"sendLatencyMillisMax", "max send latency millis"},
            {"sendMsgRate", "rate of send message"},
            {"totalAcksReceived", "total of acknowledge received"},
            {"totalBytesSent", "total of bytes sent"},
            {"totalMsgsSent", "total of messages sent"},
            {"totalSendFailed", "total of send failed"}
    };

    private static final String[] CONSUMER_LABELS = {"topic", "subscription"};
    private static final String[] PRODUCER_LABELS = {"topic", "producer_name"};

    //单例模式
    public static PulsarMetricsRecorder getInstance() {
        if (pulsarMetricsRecorder == null) {
            synchronized (PulsarMetricsRecorder.class) {
                if (pulsarMetricsRecorder == null) {
                    pulsarMetricsRecorder = new PulsarMetricsRecorder();
                    pulsarMetricsRecorder.start();
                }
            }
        }
        return pulsarMetricsRecorder;
    }

    public void registerConsumer(Consumer<?> consumer) {
        synchronized (this) {
            if (!consumerMetricsIsInit) {
                registerConsumerMetrics();
            }
        }
        if (consumer instanceof MultiTopicsConsumerImpl<?>) {
            consumersSet.addAll(((MultiTopicsConsumerImpl<?>) consumer).getConsumers());
        } else if (consumer instanceof ConsumerImpl) {
            consumersSet.add((ConsumerImpl<?>) consumer);
        }

    }

    public void registerProducer(Producer<?> producer) {
        synchronized (this) {
            if (!producerMetricsIsInit) {
                registerProducerMetrics();
            }
        }
        if (producer instanceof PartitionedProducerImpl<?>) {
            producerSet.addAll(((PartitionedProducerImpl<?>) producer).getProducers());
        } else if (producer instanceof ProducerImpl) {
            producerSet.add((ProducerImpl<?>) producer);
        }
    }

    private void registerConsumerMetrics() {
        for (String[] indicator : CONSUMER_INDICATOR) {
            Gauge.getInstance().register("pulsar_client_consumer_" + indicator[0], indicator[1], CONSUMER_LABELS);
        }
        consumerMetricsIsInit = true;
    }

    private void registerProducerMetrics() {
        for (String[] indicator : PRODUCER_INDICATOR) {
            Gauge.getInstance().register("pulsar_client_producer_" + indicator[0], indicator[1], PRODUCER_LABELS);
        }
        producerMetricsIsInit = true;
    }

    private void recordProducerMetrics() {
        producerSet.forEach(producer -> {
            ProducerStatsRecorder stats = producer.getStats();
            for (String[] indicator : PRODUCER_INDICATOR) {
                switch (indicator[0]) {
                    case "numAcksReceived":
                        Gauge.getInstance().set("pulsar_client_producer_" + indicator[0], stats.getNumAcksReceived(), producer.getTopic(), producer.getProducerName());
                        break;
                    case "numBytesSent":
                        Gauge.getInstance().set("pulsar_client_producer_" + indicator[0], stats.getNumBytesSent(), producer.getTopic(), producer.getProducerName());
                        break;
                    case "numMsgSent":
                        Gauge.getInstance().set("pulsar_client_producer_" + indicator[0], stats.getNumMsgsSent(), producer.getTopic(), producer.getProducerName());
                        break;
                    case "numSendFailed":
                        Gauge.getInstance().set("pulsar_client_producer_" + indicator[0], stats.getNumSendFailed(), producer.getTopic(), producer.getProducerName());
                        break;
                    case "sendBytesRate":
                        Gauge.getInstance().set("pulsar_client_producer_" + indicator[0], stats.getSendBytesRate(), producer.getTopic(), producer.getProducerName());
                        break;
                    case "sendLatencyMillis50pct":
                        Gauge.getInstance().set("pulsar_client_producer_" + indicator[0], stats.getSendLatencyMillis50pct() / 1000.0, producer.getTopic(), producer.getProducerName());
                        break;
                    case "sendLatencyMillis75pct":
                        Gauge.getInstance().set("pulsar_client_producer_" + indicator[0], stats.getSendLatencyMillis75pct() / 1000.0, producer.getTopic(), producer.getProducerName());
                        break;
                    case "sendLatencyMillis95pct":
                        Gauge.getInstance().set("pulsar_client_producer_" + indicator[0], stats.getSendLatencyMillis95pct() / 1000.0, producer.getTopic(), producer.getProducerName());
                        break;
                    case "sendLatencyMillis99pct":
                        Gauge.getInstance().set("pulsar_client_producer_" + indicator[0], stats.getSendLatencyMillis99pct() / 1000.0, producer.getTopic(), producer.getProducerName());
                        break;
                    case "sendLatencyMillis999pct":
                        Gauge.getInstance().set("pulsar_client_producer_" + indicator[0], stats.getSendLatencyMillis999pct() / 1000.0, producer.getTopic(), producer.getProducerName());
                        break;
                    case "sendLatencyMillisMax":
                        Gauge.getInstance().set("pulsar_client_producer_" + indicator[0], stats.getSendLatencyMillisMax() / 1000.0, producer.getTopic(), producer.getProducerName());
                        break;
                    case "sendMsgRate":
                        Gauge.getInstance().set("pulsar_client_producer_" + indicator[0], stats.getSendMsgsRate(), producer.getTopic(), producer.getProducerName());
                        break;
                    case "totalAcksReceived":
                        Gauge.getInstance().set("pulsar_client_producer_" + indicator[0], stats.getTotalAcksReceived(), producer.getTopic(), producer.getProducerName());
                        break;
                    case "totalBytesSent":
                        Gauge.getInstance().set("pulsar_client_producer_" + indicator[0], stats.getTotalBytesSent(), producer.getTopic(), producer.getProducerName());
                        break;
                    case "totalMsgsSent":
                        Gauge.getInstance().set("pulsar_client_producer_" + indicator[0], stats.getTotalMsgsSent(), producer.getTopic(), producer.getProducerName());
                        break;
                    case "totalSendFailed":
                        Gauge.getInstance().set("pulsar_client_producer_" + indicator[0], stats.getTotalSendFailed(), producer.getTopic(), producer.getProducerName());
                        break;


                }
            }
        });
    }

    private void recordConsumerMetrics() {
        consumersSet.forEach(consumer -> {
            ConsumerStatsRecorder stats = (ConsumerStatsRecorder) consumer.getStats();
            for (String[] indicator : CONSUMER_INDICATOR) {
                switch (indicator[0]) {
                    case "numAcksFailed":
                        Gauge.getInstance().set("pulsar_client_consumer_" + indicator[0], stats.getNumAcksFailed(), consumer.getTopic(), consumer.getSubscription());
                        break;
                    case "numAcksSent":
                        Gauge.getInstance().set("pulsar_client_consumer_" + indicator[0], stats.getNumAcksSent(), consumer.getTopic(), consumer.getSubscription());
                        break;
                    case "numBytesReceived":
                        Gauge.getInstance().set("pulsar_client_consumer_" + indicator[0], stats.getNumBytesReceived(), consumer.getTopic(), consumer.getSubscription());
                        break;
                    case "numMsgsReceived":
                        Gauge.getInstance().set("pulsar_client_consumer_" + indicator[0], stats.getNumMsgsReceived(), consumer.getTopic(), consumer.getSubscription());
                        break;
                    case "numReceiveFailed":
                        Gauge.getInstance().set("pulsar_client_consumer_" + indicator[0], stats.getNumReceiveFailed(), consumer.getTopic(), consumer.getSubscription());
                        break;
                    case "rateBytesReceived":
                        Gauge.getInstance().set("pulsar_client_consumer_" + indicator[0], stats.getRateBytesReceived(), consumer.getTopic(), consumer.getSubscription());
                        break;
                    case "rateMsgReceived":
                        Gauge.getInstance().set("pulsar_client_consumer_" + indicator[0], stats.getRateMsgsReceived(), consumer.getTopic(), consumer.getSubscription());
                        break;
                    case "totalAcksFailed":
                        Gauge.getInstance().set("pulsar_client_consumer_" + indicator[0], stats.getTotalAcksFailed(), consumer.getTopic(), consumer.getSubscription());
                        break;
                    case "totalAcksSent":
                        Gauge.getInstance().set("pulsar_client_consumer_" + indicator[0], stats.getTotalAcksSent(), consumer.getTopic(), consumer.getSubscription());
                        break;
                    case "totalBytesReceived":
                        Gauge.getInstance().set("pulsar_client_consumer_" + indicator[0], stats.getTotalBytesReceived(), consumer.getTopic(), consumer.getSubscription());
                        break;
                    case "totalMsgsReceived":
                        Gauge.getInstance().set("pulsar_client_consumer_" + indicator[0], stats.getTotalMsgsReceived(), consumer.getTopic(), consumer.getSubscription());
                        break;
                    case "totalReceivedFailed":
                        Gauge.getInstance().set("pulsar_client_consumer_" + indicator[0], stats.getTotalReceivedFailed(), consumer.getTopic(), consumer.getSubscription());
                        break;
                }

            }

        });


    }

    private void start() {
        this.timer = new Timer();
        this.timer.schedule(new TimerTask() {
            @Override
            public void run() {
                try {
                    recordProducerMetrics();
                    recordConsumerMetrics();
                } catch (Exception ignored) {

                }
            }
        }, statsIntervalSeconds * 1000, statsIntervalSeconds * 1000);
    }

    public static long getStatsIntervalSeconds() {
        return statsIntervalSeconds;
    }

    public static void setStatsIntervalSeconds(long statsIntervalSeconds) {
        PulsarMetricsRecorder.statsIntervalSeconds = statsIntervalSeconds;
    }

    public Set<ConsumerImpl<?>> getConsumersSet() {
        return Collections.unmodifiableSet(consumersSet);
    }

    public Set<ProducerImpl<?>> getProducerSet() {
        return Collections.unmodifiableSet(producerSet);
    }

    public boolean isProducerMetricsIsInit() {
        return producerMetricsIsInit;
    }

    public boolean isConsumerMetricsIsInit() {
        return consumerMetricsIsInit;
    }
}
